/*
 * Copyright 2015 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */
package io.netty.channel.epoll;

import io.netty.channel.Channel;
import io.netty.channel.ChannelConfig;
import io.netty.channel.ChannelMetadata;
import io.netty.channel.ChannelOutboundBuffer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.ChannelPromise;
import io.netty.channel.EventLoop;
import io.netty.channel.ServerChannel;
import io.netty.channel.unix.FileDescriptor;
import io.netty.channel.unix.Socket;

import java.net.InetSocketAddress;
import java.net.SocketAddress;

public abstract class AbstractEpollServerChannel extends AbstractEpollChannel
        implements ServerChannel
{
    private static final ChannelMetadata METADATA = new ChannelMetadata(false,
            16);

    /**
     * @deprecated Use {@link #AbstractEpollServerChannel(Socket, boolean)}.
     */
    @Deprecated
    protected AbstractEpollServerChannel(int fd)
    {
        this(new Socket(fd), false);
    }

    /**
     * @deprecated Use {@link #AbstractEpollServerChannel(Socket, boolean)}.
     */
    @Deprecated
    protected AbstractEpollServerChannel(FileDescriptor fd)
    {
        this(new Socket(fd.intValue()));
    }

    /**
     * @deprecated Use {@link #AbstractEpollServerChannel(Socket, boolean)}.
     */
    @Deprecated
    protected AbstractEpollServerChannel(Socket fd)
    {
        this(fd, isSoErrorZero(fd));
    }

    protected AbstractEpollServerChannel(Socket fd, boolean active)
    {
        super(null, fd, Native.EPOLLIN, active);
    }

    @Override
    public ChannelMetadata metadata()
    {
        return METADATA;
    }

    @Override
    protected boolean isCompatible(EventLoop loop)
    {
        return loop instanceof EpollEventLoop;
    }

    @Override
    protected InetSocketAddress remoteAddress0()
    {
        return null;
    }

    @Override
    protected AbstractEpollUnsafe newUnsafe()
    {
        return new EpollServerSocketUnsafe();
    }

    @Override
    protected void doWrite(ChannelOutboundBuffer in) throws Exception
    {
        throw new UnsupportedOperationException();
    }

    @Override
    protected Object filterOutboundMessage(Object msg) throws Exception
    {
        throw new UnsupportedOperationException();
    }

    abstract Channel newChildChannel(int fd, byte[] remote, int offset, int len)
            throws Exception;

    final class EpollServerSocketUnsafe extends AbstractEpollUnsafe
    {
        // Will hold the remote address after accept(...) was successful.
        // We need 24 bytes for the address as maximum + 1 byte for storing the
        // length.
        // So use 26 bytes as it's a power of two.
        private final byte[] acceptedAddress = new byte[26];

        @Override
        public void connect(SocketAddress socketAddress,
                SocketAddress socketAddress2, ChannelPromise channelPromise)
        {
            // Connect not supported by ServerChannel implementations
            channelPromise.setFailure(new UnsupportedOperationException());
        }

        @Override
        void epollInReady()
        {
            assert eventLoop().inEventLoop();
            final ChannelConfig config = config();
            if (shouldBreakEpollInReady(config))
            {
                clearEpollIn0();
                return;
            }
            final EpollRecvByteAllocatorHandle allocHandle = recvBufAllocHandle();
            allocHandle.edgeTriggered(isFlagSet(Native.EPOLLET));

            final ChannelPipeline pipeline = pipeline();
            allocHandle.reset(config);
            epollInBefore();

            Throwable exception = null;
            try
            {
                try
                {
                    do
                    {
                        // lastBytesRead represents the fd. We use lastBytesRead
                        // because it must be set so that the
                        // EpollRecvByteAllocatorHandle knows if it should try
                        // to read again or not when autoRead is
                        // enabled.
                        allocHandle.lastBytesRead(fd().accept(acceptedAddress));
                        if (allocHandle.lastBytesRead() == -1)
                        {
                            // this means everything was handled for now
                            break;
                        }
                        allocHandle.incMessagesRead(1);

                        int len = acceptedAddress[0];
                        readPending = false;
                        pipeline.fireChannelRead(
                                newChildChannel(allocHandle.lastBytesRead(),
                                        acceptedAddress, 1, len));
                    }
                    while (allocHandle.continueReading());
                }
                catch (Throwable t)
                {
                    exception = t;
                }
                allocHandle.readComplete();
                pipeline.fireChannelReadComplete();

                if (exception != null)
                {
                    pipeline.fireExceptionCaught(exception);
                }
            }
            finally
            {
                epollInFinally(config);
            }
        }
    }
}
